package org.apache.hadoop.hdfs.server.namenode;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.DF;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.server.common.Util;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Collections2;
import com.google.common.base.Predicate;

/**
 * 提供了一个hasAvailableDiskSpace()方法，
 * 这个方法仅仅在NameNode有足够的磁盘空间满足所有磁盘使用需求的时候 返回true，包括文件系统的edits dir，包括配置文件内设置的dirs。
 * 否则，返回false.
 *
 * 从NameNodeResourceChecker这个类的注释上可以清楚的得知，就是为了检查配置里的路径的磁盘空间大小是否满足使用需求的。
 * 它最核心的两个变量一个是duReserved，一个是volumes。
 * 1. duReserved,
 *     这是一个long型变量，通过duReserved = conf.getLong(DFSConfigKeys.DFS_NAMENODE_DU_RESERVED_KEY,...);这行代码，
 * 　　 可以溯源到hdfs-default.xml中的dfs.namenode.resource.du.reserved这个配置选项，配置文件中默认为空，但是getLong()方法有一个默认大小，是100M。
 *     那些待检查空间大小的路径首先会同这个duReserved进行比较，小于duReserved的话，返回false。
 *
 *     返回false结果会导致什么情况出现呢？
 *     通过跟踪hasAvailableDiskSpace()方法的被调用代码，一步一步进行梳理，最终会 进入到FSNamesystem类的一个内部线程类中：NameNodeResourceMonitor，
 *     这个线程会周期性的调用NameNodeResourceChecker的hasAvailableRecourses()方法进行磁盘资源检查，
 *     如果一旦发现有资源不足的情况，会使NameNode进入安全模式。如果随后返回的状态代表资源大小到达可使用的级别，那么这个线程就使NameNode退出安全模式。
 * 2. volumes,
 *     这是一个map，通过addDirToCheck()方法及注释可知，主要是用来存放需要进行磁盘空间检查的dirs。
 *     然后在hasAvailableDiskSpace()中将这个map传入到了真正负责磁盘检查逻辑的类：NameNodeResourcePolicy，这个类就是用来遍历volumes这个map进行资源检查的。
 */
@InterfaceAudience.Private
public class NameNodeResourceChecker {
  private static final Log LOG = LogFactory.getLog(NameNodeResourceChecker.class.getName());

  // 配置文件内的每个被配置的磁盘路径空间的默认大小
  private final long duReserved;

  private final Configuration conf;
  // 一个map，用来存放所有应该检查的配置路径，在NameNodeResourceChecker实例化的时候进行初始化
  private Map<String, CheckedVolume> volumes;
  private int minimumRedundantVolumes;
  
  @VisibleForTesting
  class CheckedVolume implements CheckableNameNodeResource {
    private DF df;
    private boolean required;
    private String volume;
    
    public CheckedVolume(File dirToCheck, boolean required) throws IOException {
      df = new DF(dirToCheck, conf);
      this.required = required;
      volume = df.getFilesystem();
    }
    
    public String getVolume() {
      return volume;
    }
    
    @Override
    public boolean isRequired() {
      return required;
    }

    @Override
    public boolean isResourceAvailable() {
      long availableSpace = df.getAvailable();
      // duReserved就是默认配置好的最小需要的磁盘空间
      // duReserved空间默认是100Mb，默认情况下edits目录起码要有100Mb空间来写入日志，
      // 否则就是检查资源失败
      if (availableSpace < duReserved) {
        LOG.warn("Space available on volume '" + volume + "' is " + availableSpace + ", which is below the configured reserved amount " + duReserved);
        return false;
      } else {
        return true;
      }
    }
    
    @Override
    public String toString() {
      return "volume: " + volume + " required: " + required + " resource available: " + isResourceAvailable();
    }
  }

  /**
   * 构造方法。
   * 创建一个NameNodeResourceChecker对象，初始化volumes Map，用来存放需要进行磁盘空间检查的路径。
   * 通过将这个volumes的map传递给NameNodeResourcePolicy类进行遍历volumes Map来进行检查edits dirs
   * 和一些配置文件中配置的dirs。
   */
  public NameNodeResourceChecker(Configuration conf) throws IOException {
    this.conf = conf;
    volumes = new HashMap<String, CheckedVolume>();

    // 从配置文件中获取DFS_NAMENODE_DU_RESERVED_KEY配置， 默认100MB
    duReserved = conf.getLong(DFSConfigKeys.DFS_NAMENODE_DU_RESERVED_KEY, DFSConfigKeys.DFS_NAMENODE_DU_RESERVED_DEFAULT);
    
    Collection<URI> extraCheckedVolumes = Util.stringCollectionAsURIs(conf.getTrimmedStringCollection(DFSConfigKeys.DFS_NAMENODE_CHECKED_VOLUMES_KEY));
    
    Collection<URI> localEditDirs = Collections2.filter(
        FSNamesystem.getNamespaceEditsDirs(conf),
        new Predicate<URI>() {
          @Override
          public boolean apply(URI input) {
            if (input.getScheme().equals(NNStorage.LOCAL_URI_SCHEME)) {
              return true;
            }
            return false;
          }
        });

    // Add all the local edits dirs, marking some as required if they are configured as such.
    for (URI editsDirToCheck : localEditDirs) {
      addDirToCheck(editsDirToCheck, FSNamesystem.getRequiredNamespaceEditsDirs(conf).contains( editsDirToCheck));
    }

    // All extra checked volumes are marked "required"
    for (URI extraDirToCheck : extraCheckedVolumes) {
      addDirToCheck(extraDirToCheck, true);
    }
    
    minimumRedundantVolumes = conf.getInt(
        DFSConfigKeys.DFS_NAMENODE_CHECKED_VOLUMES_MINIMUM_KEY,
        DFSConfigKeys.DFS_NAMENODE_CHECKED_VOLUMES_MINIMUM_DEFAULT);
  }

  /**
   * Add the volume of the passed-in directory to the list of volumes to check.
   * If <code>required</code> is true, and this volume is already present, but
   * is marked redundant, it will be marked required. If the volume is already
   * present but marked required then this method is a no-op.
   * 
   * @param directoryToCheck The directory whose volume will be checked for available space.
   */
  private void addDirToCheck(URI directoryToCheck, boolean required) throws IOException {
    File dir = new File(directoryToCheck.getPath());
    if (!dir.exists()) {
      throw new IOException("Missing directory "+dir.getAbsolutePath());
    }
    
    CheckedVolume newVolume = new CheckedVolume(dir, required);
    CheckedVolume volume = volumes.get(newVolume.getVolume());
    if (volume == null || !volume.isRequired()) {
      volumes.put(newVolume.getVolume(), newVolume);
    }
  }

  /**
   * 如果磁盘空间满足所有配置要求的大小并且至少满足一个配置的冗余大小就返回true
   */
  public boolean hasAvailableDiskSpace() {
    return NameNodeResourcePolicy.areResourcesAvailable(volumes.values(), minimumRedundantVolumes);
  }

  /**
   * Return the set of directories which are low on space.
   * @return the set of directories whose free space is below the threshold.
   */
  @VisibleForTesting
  Collection<String> getVolumesLowOnSpace() throws IOException {
    Collection<String> lowVolumes = new ArrayList<String>();
    for (CheckedVolume volume : volumes.values()) {
      lowVolumes.add(volume.getVolume());
    }
    return lowVolumes;
  }
  
  @VisibleForTesting
  void setVolumes(Map<String, CheckedVolume> volumes) {
    this.volumes = volumes;
  }
  
  @VisibleForTesting
  void setMinimumReduntdantVolumes(int minimumRedundantVolumes) {
    this.minimumRedundantVolumes = minimumRedundantVolumes;
  }
}
